package com.yanggu.flink.datastream_api.sink

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.connector.jdbc.{JdbcConnectionOptions, JdbcExecutionOptions, JdbcSink, JdbcStatementBuilder}
import org.apache.flink.streaming.api.scala._
import com.yanggu.flink.datastream_api.pojo.Book

import java.sql.PreparedStatement

/**
 * 将数据输出到mysql中, 使用官方提供的sink
 * https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/connectors/datastream/jdbc/
 */
object JdbcSinkDemo01 {

  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment

    val dataStream = environment
      .fromCollection(
        Seq(Book(101L, "Stream Processing with Apache Flink", "Fabian Hueske, Vasiliki Kalavri", 2019),
          Book(102L, "Streaming Systems", "Tyler Akidau, Slava Chernyak, Reuven Lax", 2018),
          Book(103L, "Designing Data-Intensive Applications", "Martin Kleppmann", 2017),
          Book(104L, "Kafka: The Definitive Guide", "Gwen Shapira, Neha Narkhede, Todd Palino", 2017))
      )

    val sql = "INSERT INTO books (id, title, authors, year) VALUES (?, ?, ?, ?)"

    val function = new JdbcStatementBuilder[Book] {
      override def accept(statement: PreparedStatement, book: Book): Unit = {
        statement.setLong(1, book.id)
        statement.setString(2, book.title)
        statement.setString(3, book.authors)
        statement.setInt(4, book.year)
      }
    }

    //JDBC的sink是批处理, 为了提高运行效率。(当数据攒满多少个，或者超过多少时间就输出)
    val executionOptions = JdbcExecutionOptions.builder()
      .withBatchSize(1000)
      .withBatchIntervalMs(200)
      .withMaxRetries(5)
      .build()

    val tool = ParameterTool.fromPropertiesFile(getClass.getResourceAsStream("/jdbc.properties"))

    val connectionOptions = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
      .withUrl(tool.get("jdbc.url"))
      .withDriverName(tool.get("jdbc.driverClassName"))
      .withUsername(tool.get("jdbc.username"))
      .withPassword(tool.get("jdbc.password"))
      .build()

    val jdbcSinkFunction = JdbcSink.sink[Book](sql, function, executionOptions, connectionOptions)

    dataStream.addSink(jdbcSinkFunction)

    environment.execute("JdbcSinkDemo01 Job")
  }

}
